package com.teamsun.mqtt.handle;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;

import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;

import com.teamsun.mqtt.entity.MsgRep;
import com.teamsun.mqtt.entity.SendMsg;
import com.teamsun.mqtt.entity.vo.SendableMsg;
import com.teamsun.mqtt.service.LogService;
import com.teamsun.mqtt.service.MessagePushService;
import com.teamsun.mqtt.service.MsgService;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttPubAckMessage;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.util.AttributeKey;
import io.netty.util.ReferenceCountUtil;

/**
 * 发布消息处理类
 * 
 * @author acer
 *
 */
@Sharable
public class PushServiceHandle extends ChannelInboundHandlerAdapter {

	Logger logger = Logger.getLogger(getClass());

	/**
	 * 用于根据登录的客户端标识找channel
	 */
	ConcurrentHashMap<String, Channel> str2channel;
	/**
	 * 用于根channel 找登录的客户端
	 */
	ConcurrentHashMap<Channel, String> channel2str;
	/**
	 * 消息 队列
	 */
	BlockingQueue<MsgRep> msgRepsque;


	/**
	 * 数据库操作线程组
	 */
	ExecutorService dboptgroup;


	@Autowired
	MessagePushService messagePushService;


	
	@Autowired
	MsgService msgService;



	public PushServiceHandle(ExecutorService executorService) {
		this.dboptgroup = executorService;
	}

	/**
	 * 业务开始
	 */
	public void channelRead(final ChannelHandlerContext ctx, Object msg) {

		if (msg instanceof MqttMessage) {

			MqttMessage message = (MqttMessage) msg;
			MqttMessageType messageType = message.fixedHeader().messageType();

			switch (messageType) {
			case PUBLISH:// 客户端发布普通消息
				MqttPublishMessage messagepub = (MqttPublishMessage) msg;
				pub(ctx, messagepub);
				break;

			case PUBREL: // 客户端发布释放
				pubrel(ctx, message);
				break;
			case PUBREC:// 客户端发布收到
				pubrec(ctx, message);
				break;
			case PUBCOMP:
			case PUBACK:
				ReferenceCountUtil.release(message);
				break;
			default:
				ctx.fireChannelRead(msg);
				break;
			}

		}
		
		else
			ctx.channel().close();
	}

	/**
	 * 处理客户端的发布请求 根据客户端请求的QOS级别 发送相应的响应 把相应的消息存储在数据库 现在的 客户端标识是 发送方的
	 * 
	 * @param ctx
	 * @param messagepub
	 */
	private void pub(final ChannelHandlerContext ctx, MqttPublishMessage messagepub) {

		MqttQoS mqttQoS = messagepub.fixedHeader().qosLevel();

		MqttFixedHeader fixedHeader = null;
		if (mqttQoS.value() <= 1) {
			// 不是级别最高的QOS 返回 puback 即可
			fixedHeader = new MqttFixedHeader(MqttMessageType.PUBACK, false, mqttQoS, false, 0);
		} else {
			// 否则发送发布收到 QOS级别会2
			fixedHeader = new MqttFixedHeader(MqttMessageType.PUBREC, false, MqttQoS.EXACTLY_ONCE, false, 0);
		}

		int messageid = messagepub.variableHeader().messageId();
		MqttMessageIdVariableHeader connectVariableHeader = MqttMessageIdVariableHeader.from(messageid);

		MqttPubAckMessage ackMessage = new MqttPubAckMessage(fixedHeader, connectVariableHeader);
		ctx.write(ackMessage);

		final String iden = channel2str.get(ctx.channel());
		final String topname = messagepub.variableHeader().topicName();
		MsgRep msgRep = new MsgRep(messageid, topname, iden, messagepub.content());
		if (!msgRepsque.offer(msgRep)) {
			List<MsgRep> msgReps = new ArrayList<MsgRep>();
			msgRepsque.drainTo(msgReps);
			msgReps.add(msgRep);
			subClearSaveMsg(msgReps);
		}
		
		

		SendableMsg sendableMsg = new SendableMsg();
		sendableMsg.setDupTimes(0);
		sendableMsg.setMessageid(msgRep.getId());
		sendableMsg.setMsgContent(messagepub.content());
		sendableMsg.setQoS(mqttQoS);
		sendableMsg.setSendclientid(iden);
		sendableMsg.setTopname(topname);

		messagePushService.sendPubMsg(sendableMsg);
		// 如果不能放进去了 就清除并且提交到数据库
		

		LogService.log(PushServiceHandle.this.logger, Level.INFO, iden + "发布消息，主题为" + topname);
	}

	/**
	 * 处理客户端过来的发布释放 在消息池里根据客户端标识和消息id面拿到一个消息 并且把这个消息发送出去 现在的 客户端标识是 发送方的
	 */
	private void pubrel(final ChannelHandlerContext ctx, MqttMessage messagepub) {

		MqttMessageIdVariableHeader variableHeader = (MqttMessageIdVariableHeader) messagepub.variableHeader();

		Integer messageid = variableHeader.messageId();

		MqttQoS mqttQoS = messagepub.fixedHeader().qosLevel();

		MqttFixedHeader fixedHeader = null;

		if (mqttQoS.value() <= 1)
			// 发送发布收到 QOS级别会1
			fixedHeader = new MqttFixedHeader(MqttMessageType.PUBCOMP, false, MqttQoS.AT_MOST_ONCE, false, 0);
		else
			// 否则发送发布收到 QOS级别会2
			fixedHeader = new MqttFixedHeader(MqttMessageType.PUBCOMP, false, MqttQoS.EXACTLY_ONCE, false, 0);

		MqttMessageIdVariableHeader connectVariableHeader = MqttMessageIdVariableHeader.from(messageid);

		MqttPubAckMessage ackMessage = new MqttPubAckMessage(fixedHeader, connectVariableHeader);
		ctx.write(ackMessage);

	}

	/**
	 * 处理客户端 发布收到
	 * 
	 * 对客户端发送发布释放 根据 客户端收到的messageid 找到相应的message 并且 存储到消息记录里面 现在的 客户端标识是 接受方的
	 * 现在的messageid是数据库里面的主键id 最后移除重发队列 防止消息重发
	 * 
	 * @param ctx
	 * @param messagepub
	 */
	private void pubrec(final ChannelHandlerContext ctx, MqttMessage messagepub) {

		MqttMessageIdVariableHeader variableHeader = (MqttMessageIdVariableHeader) messagepub.variableHeader();

		final Integer messageid = variableHeader.messageId();

		MqttQoS mqttQoS = messagepub.fixedHeader().qosLevel();

		MqttFixedHeader fixedHeader = null;

		if (mqttQoS != MqttQoS.EXACTLY_ONCE)
			// 发送发布收到 QOS级别会1
			fixedHeader = new MqttFixedHeader(MqttMessageType.PUBREL, false, MqttQoS.AT_MOST_ONCE, false, 0);
		else
			// 否则发送发布收到 QOS级别会2
			fixedHeader = new MqttFixedHeader(MqttMessageType.PUBREL, false, MqttQoS.EXACTLY_ONCE, false, 0);

		MqttMessageIdVariableHeader connectVariableHeader = MqttMessageIdVariableHeader.from(messageid);

		MqttPubAckMessage ackMessage = new MqttPubAckMessage(fixedHeader, connectVariableHeader);
		ctx.write(ackMessage);
		

		AttributeKey<Boolean>  attributeKey=AttributeKey.valueOf("recv:"+messageid);
		ctx.channel().attr(attributeKey).set(true);


	}

	
	/**
	 * 保持消息 将多余的消息放入数据库
	 * 
	 * @param msgReps
	 */
	private void subClearSaveMsg(final Collection<MsgRep> msgReps) {

		dboptgroup.submit(()->{
			for (MsgRep msgRep : msgReps) {
				msgService.saveMsgRep(msgRep);
			}
		});
	}

	@Override
	public void channelInactive(final ChannelHandlerContext ctx) throws Exception {
		super.channelInactive(ctx);
		ctx.close();
	}

	public void setStr2channel(ConcurrentHashMap<String, Channel> str2channel) {
		this.str2channel = str2channel;
	}

	public void setChannel2str(ConcurrentHashMap<Channel, String> channel2str) {
		this.channel2str = channel2str;
	}

	public void setMsgRepsque(BlockingQueue<MsgRep> msgRepsque) {
		this.msgRepsque = msgRepsque;
	}

	
	
	

}
